#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include "thrdpool.h"
#include "list.h"

#define lock(t)       pthread_mutex_lock((&t))
#define trylock(t)    pthread_mutex_trylock((&t))
#define unlock(t)     pthread_mutex_unlock((&t))
#define waite(t, c)   pthread_cond_wait((&t), (&c))
#define notify_one(c) pthread_cond_signal((&c))

typedef struct {
    struct list_head node;
    Task             task;
} QueueNode;

struct thrdqueue_t {
    struct list_head head;
    pthread_t        pid;
    pthread_mutex_t  mutex;
    pthread_cond_t   cond;
    unsigned int     waiting_couts; //等待处理的任务数

    Thrd* pool;
};

struct thrdpool_t {
    pthread_t* pid;
    Queue*     queues;
    int        thread_couts;
    int        queues_couts; //由线程数决定，1个线程对应1个队列
    int        is_working;
};

static int _queue_init(Thrd* pool)
{
    pool->queues =
        (Queue*)calloc(1, sizeof(struct thrdqueue_t) * pool->queues_couts);
    if(NULL == pool->queues) { return 1; }
    for(int i = 0; i < pool->queues_couts; ++i) {
        pthread_mutex_init(&pool->queues[i].mutex, NULL);
        pthread_cond_init(&pool->queues[i].cond, NULL);
        INIT_LIST_HEAD(&pool->queues[i].head);
        pool->queues[i].pool          = pool;
        pool->queues[i].waiting_couts = 0;
    }
    return 0;
}

static void _queue_clean(Queue* task_queue)
{
    QueueNode *pos, *n;
    list_for_each_entry_safe(pos, n, &task_queue->head, node)
    {
        if(pos) {
            free(pos);
            pos = NULL;
        }
    }
}

static void _queue_deinit(Thrd* pool)
{
    for(int i = 0; i < pool->queues_couts; ++i) {
        _queue_clean(&pool->queues[i]);
    }
}
static int _queue_push(Queue* queue, Task* new_task)
{
    QueueNode* new_node = (QueueNode*)calloc(1, sizeof(QueueNode));
    if(NULL == new_node) { return -1; }

    memcpy(&new_node->task, new_task, sizeof(Task));
    lock(queue->mutex);
    list_add_tail(&new_node->node, &queue->head);
    ++queue->waiting_couts;
    notify_one(queue->cond);
    unlock(queue->mutex);

    return 0;
}

static int _queue_pop(Queue* queue, Task* task)
{
    lock(queue->mutex);
    if(queue->waiting_couts == 0) { return 1; }
    QueueNode* node = list_first_entry(&queue->head, QueueNode, node);
    list_del(&node->node);
    --queue->waiting_couts;
    unlock(queue->mutex);
    memcpy(task, &node->task, sizeof(Task));
    free(node);

    return 0;
}

void* thrd_func(void* arg)
{
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
    Queue* task_queue = (Queue*)arg;
    Thrd*  pool       = task_queue->pool;
    while(pool->is_working) {
        // while(list_empty(&task_queue->head))//同下
        while(task_queue->waiting_couts == 0) {
            waite(task_queue->cond, task_queue->mutex);
            unlock(task_queue->mutex);
        }
        Task task;
        //出队

        if(_queue_pop(task_queue, &task)) { continue; }
        task.handle(task.ctx, task.event);
    }
    return (void*)0;
}

static int _thrds_init(Thrd* pool)
{
    //线程初始化
    pool->pid = (pthread_t*)calloc(1, sizeof(pthread_t) * pool->thread_couts);
    if(NULL == pool->pid) { return 1; }
    int real_cout = 0;
    for(int i = 0; i < pool->thread_couts; ++i) {
        if(pthread_create(&pool->pid[i], NULL, thrd_func, &pool->queues[i])) {
            continue;
        }
        ++real_cout;
    }
    if(real_cout != pool->thread_couts) {
        for(int i = 0; i < real_cout; ++i) { pthread_cancel(pool->pid[i]); }
        return 1;
    }
    return 0;
}

static void _thrds_end(Thrd* pool)
{
    for(int i = 0; i < pool->thread_couts; ++i) {
        notify_one(pool->queues[i].cond);
        pthread_cancel(pool->pid[i]);
    }
    for(int i = 0; i < pool->thread_couts; ++i) {
        pthread_join(pool->pid[i], NULL);
    }
    return;
}

static void _thrd_deinit(Thrd* pool)
{
    if(pool->pid) {
        _thrds_end(pool);
        free(pool->pid);
        pool->pid = NULL;
    }
}

Thrd* thrdpool_init(int thrdnums)
{
    Thrd* pool = (Thrd*)calloc(1, sizeof(struct thrdpool_t));
    if(NULL == pool) { return NULL; }

    pool->thread_couts = thrdnums;
    pool->queues_couts = thrdnums;
    pool->is_working   = 1;

    if(_queue_init(pool)) {
        free(pool->pid);
        goto queue_err;
    }

    if(_thrds_init(pool)) { goto thrd_err; }

    for(int i = 0; i < pool->thread_couts; ++i) {
        pool->queues[i].pid = pool->pid[i];
    }
    return pool;

thrd_err:
    if(pool->pid) { free(pool->pid); }
queue_err:
    _queue_deinit(pool);
    free(pool);
    return NULL;
}

static int add_index = -1;

int thrdpool_add(Thrd* pool, Task* task)
{
    /*如果是普通任务，轮询即可*/
    add_index = (add_index + 1) % pool->queues_couts;
    // printf("add_index:%d\n",add_index);
    return _queue_push(&pool->queues[add_index], task);
    /*如果是网络io，需要注意确保关闭socket后不在有对该socket的读写操作*/
}

void thrdpool_destory(Thrd* pool)
{
    pool->is_working = 0;
    _thrd_deinit(pool);
    _queue_deinit(pool);
}
